/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.service.qos.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.Vector;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.matching.Profile;
import cgl.narada.matching.ProfileRequest;
import cgl.narada.protocol.DataReceived;
import cgl.narada.protocol.Handler;
import cgl.narada.protocol.Protocol;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.ServicePropertiesFactory;
import cgl.narada.service.client.impl.ClientServiceImpl;
import cgl.narada.transport.TransmissionManager;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;
import cgl.narada.transport.TransportHandlerImpl;

/**
 * This class is responsible for managing communications between the broker and
 * entities
 * 
 * @author Shrideep Pallickara $Date$ $Revision$
 */

public class CommunicationsService extends Thread implements Handler {

    /** Deals with the transport related aspects of communication */
    private TransportHandler transportHandler;

    /** Provides information about ongoing transmissions. */
    private TransmissionManager transmissionManager;

    /** The broker to which the client is attached to */
    private String gesBroker;

    private int packetNumber = 0;

    private QosServiceImpl qosServiceImpl;

    private boolean debug = false;

    private Vector receivedEvents;

    private Object syncObject;

    private boolean waiting = true;

    private boolean keepLooping = true;

    private String moduleName = "CommunicationsService: ";

    public CommunicationsService(QosServiceImpl qosServiceImpl) {
        syncObject = new Object();
        receivedEvents = new Vector();
        transportHandler = new TransportHandlerImpl(this);
        transmissionManager = transportHandler.getTransmissionManager();
        this.qosServiceImpl = qosServiceImpl;
        setName("CommunicationsService");
    }

    /** This method indicates that the node in question is a client */
    public boolean isClient() {
        return true;
    }

    private void oldHandleData(DataReceived _dataReceived) {
        receivedEvents.addElement(_dataReceived);
        Thread.currentThread().yield();
        if (waiting && (receivedEvents.size() > 0)) {
            synchronized (syncObject) {
                syncObject.notify();
            }
            ;
        }
    }

    public void handleData(byte[] data, String sender) {
        processData(data, sender);
    }

    /** Terminating the service */
    public void terminateService() throws ServiceException {
        keepLooping = false;
        synchronized (syncObject) {
            syncObject.notify();
        }
        ;
    }

    public void run() {
        try {
            while (keepLooping) {
                if (receivedEvents.size() == 0) {
                    synchronized (syncObject) {
                        waiting = true;
                        if (debug) {
                            System.out.println(moduleName
                                + "Waiting for notification");
                        }

                        syncObject.wait();

                        if (debug)
                            System.out.println(moduleName + "Waking up!");
                        waiting = false;
                    }
                    ;/* end synchronized(syncObject) */
                }/* end if receivedEvents.size() == 0 */

                if (receivedEvents.size() == 0) {
                    continue;
                }

                DataReceived dataToProcess = (DataReceived) receivedEvents
                    .elementAt(0);
                // processData(dataToProcess);
                receivedEvents.removeElementAt(0);

            }/* end (while) */
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * This method is responsible for dealing with packets of messages that are
     * received over a link, based on the protocol headers that are contained
     * within the packets. The DataReceived object is the encapsulation of the
     * data that is received by one of the ReceiverThreads within the transport
     * layer.
     * 
     * @param _dataReceived
     *            The data that is received by one of the ReceiverThreads
     */
    public void processData(byte[] data, String sender) {

        if (data.length == 0) {
            System.out.println(moduleName + "handleData()-> length == 0");
            return;
        }

        int dataInfo = (int) data[0];
        packetNumber++;

        switch (dataInfo) {
        case (int) Protocol.DATA:
            processDataPackets(data);
            break;

        default:
            System.out.println(moduleName + "Error::Data[1] ->" + dataInfo
                + " Data length = " + data.length);
            break;
        }/* end switch-case */

    }

    private void processDataPackets(byte[] dataBytes) {
        int dataInfo = dataBytes[1];

        switch (dataInfo) {
        case (int) Protocol.NB_EVENT:
            if (debug) {
                System.out.println(moduleName + "Received NB_EVENT header ");
            }
            NBEvent nbEvent = cgl.narada.event.impl.NBEventGenerator
                .unmarshallEvent(dataBytes);
            long startTime = System.currentTimeMillis();
            qosServiceImpl.processReceivedEvent(nbEvent);
            long delay = System.currentTimeMillis() - startTime;
            // System.out.println(moduleName + "Processing time = " + delay);
            break;

        case (int) Protocol.NODE_ADDRESS:
            processNodeAddressResponse(dataBytes);
            break;

        // HG: A security issue... No access allowed... check message...
        case (int) Protocol.CLIENT_SETUP_FAILURE:
            processClientSetupFailure(dataBytes);
            break;
        // ---
        default:
            System.out.println(moduleName + "Error::Data[1] ->" + dataInfo
                + " Data length = " + dataBytes.length);
            break;
        }/* end switch-case */

    }

    // HG: Client setup failure
    private void processClientSetupFailure(byte[] msg) {
        // Get the message
        byte[] msgBytes = new byte[msg.length - 1];
        System.arraycopy(msg, 1, msgBytes, 0, msgBytes.length);
        System.out.println("ERROR: " + new String(msgBytes));
    }

    // --

    // HG: Mgmt

    // Process Failure to Broker Connection...
    public void processConnectionLost() {
        qosServiceImpl.processConnectionLoss();
    }

    // --

    private void processNodeAddressResponse(byte[] addressResponse) {
        ByteArrayInputStream baInputStream = new ByteArrayInputStream(
                                                                      addressResponse);
        DataInputStream din = new DataInputStream(baInputStream);

        try {
            din.readByte();
            /** Protocol.DATA */
            din.readByte();
            /** Protocol.NODE_ADDRESS */

            int indicator = din.readByte();
            /** Protocol.CLIENT_SETUP */
            if (indicator == Protocol.CLIENT_SETUP_SUCCESS) {
                if (debug) {
                    System.out.println(moduleName
                        + "Client registration success!");
                }
                qosServiceImpl.connectionInitialized();
            } else {
                System.out.println(moduleName
                    + "Problem with client registration");
            }
        } catch (IOException ioe) {
            System.out.println(moduleName + "Error marshalling stream" + ioe);
        } /* end try-catch */

    }

    public void loadCommunicationsOfType(Properties props, String commType)
        throws ServiceException {
        try {
            transportHandler.loadCommunicationsOfType(props, commType);
        } catch (TransportException transEx) {
            throw new ServiceException(moduleName + transEx.toString());
        }
    }

    /**
     * This method is responsible for registering an entity with a specified
     * broker
     */
    public void registerEntity(int entityId) throws ServiceException {
        System.out.println(moduleName + "Registering [" + entityId + "]");
        if (gesBroker == null) {
            throw new ServiceException(moduleName
                + "Connection to broker not initialized ");
        }

        ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
        DataOutputStream dout = new DataOutputStream(baOutputStream);

        try {
            dout.writeByte(Protocol.DATA);
            dout.writeByte(Protocol.NODE_ADDRESS);
            dout.writeByte(Protocol.CLIENT_SETUP);
            dout.writeInt(entityId);

        } catch (IOException ioe) {
            throw new ServiceException(moduleName + "Error marshalling stream"
                + ioe);
        } /* end try-catch */
        sendData(baOutputStream.toByteArray());

    }

    public String setupLink(Properties props, String linkType)
        throws ServiceException {
        gesBroker = null;
        try {
            gesBroker = transportHandler.setupLink(props, linkType);
        } catch (TransportException transEx) {
            throw new ServiceException(moduleName + transEx.toString());
        }
        return gesBroker;
    }

    public void closeConnection() throws ServiceException {
        while (hasPendingTransfers()) {
            try {
                sleep(100);
            } catch (InterruptedException e) {
                System.out
                    .println(moduleName
                        + "Problems sleeping while waiting for transfer completion");
            }
        }
        System.out.println(moduleName + "Closing Link => " + gesBroker);
        transportHandler.closeLinks(gesBroker);
        transportHandler.shutdown();
    }

    /** Checks to see if there are pending transfers. */
    public boolean hasPendingTransfers() {
        return transmissionManager.hasPendingTransfers();
    }

    public void sendData(byte[] dataToBeSent) throws ServiceException {
        if (gesBroker == null) {
            throw new ServiceException(moduleName
                + "Connection to broker not initialized ");
        }

        try {
            transportHandler.sendData(dataToBeSent, gesBroker);
        } catch (TransportException transEx) {
            throw new ServiceException(moduleName + transEx.toString());
        }
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        /** These properties pertain to setting up a TCP link */
        props.put("hostname", args[0]);
        props.put("portnum", args[1]);

        int entityId = 76543210;
        String configFileLocation = "config/ServiceProperties.properties";

        try {
            ServicePropertiesFactory servicePropertiesFactory = ServicePropertiesFactory
                .getInstance();

            servicePropertiesFactory
                .intializeServiceProperties(entityId, configFileLocation);
            ServiceProperties serviceProperties = servicePropertiesFactory
                .getServiceProperties(entityId);
            QosServiceImpl qosService = new QosServiceImpl(serviceProperties);
            CommunicationsService commService = new CommunicationsService(
                                                                          qosService);

            ClientServiceImpl client = new ClientServiceImpl(entityId,
                                                             qosService);

            commService.loadCommunicationsOfType(props, "niotcp");
            commService.setupLink(props, "niotcp");
            commService.registerEntity(entityId);

            Profile profile = client
                .createProfile(TemplateProfileAndSynopsisTypes.STRING,
                               "test/shrideep");
            ProfileRequest request = new ProfileRequest(profile, true);
            commService.sendData(request.getBytes());

            NBEventGenerator generator = new NBEventGenerator();
            NBEvent nbEvent = generator
                .generateEvent(123, entityId, false, true, false, false, 0,
                               true, false, null, false, 0,
                               TemplateProfileAndSynopsisTypes.STRING,
                               "test/stuff",
                               new String("Hey I am sending you a message")
                                   .getBytes());
            commService.sendData(nbEvent.getBytes());
        } catch (ServiceException serEx) {
            System.out.println(serEx);
        }

    }

}
